package com.atguigu.interceptor;

import java.util.ArrayList;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.List;
import java.util.Map;

/**
 * @author fuc
 * @create 2020-05-31
 */
public class TypeInterceptor implements Interceptor {
  /** 声明一个存放事件的集合 */
  private List<Event> addHeadersEvents;

  @Override
  public void initialize() {
    addHeadersEvents = new ArrayList<>();
  }

  /** 单个事件拦截 */
  @Override
  public Event intercept(Event event) {
    // 1.获取事件中的头信息
    Map<String, String> headers = event.getHeaders();

    // 2.获取事件中的body信息
    String body = new String(event.getBody());

    // 3.根据body中是否有“hello”来决定添加怎样的头信息
    if (body.contains("hello")) {
      // 4.添加头信息
      headers.put("type", "atguigu");
    } else {
      // 4.添加头信息
      headers.put("type", "bigdata");
    }

    return event;
  }

  /** 批量拦截 */
  @Override
  public List<Event> intercept(List<Event> events) {
    // 1.清空集合
    addHeadersEvents.clear();
    // 2.遍历Events
    events.stream()
        .forEach(
            // 3.给每一个事件添加头信息
            event -> addHeadersEvents.add(event));
    // 4.返回结果
    return addHeadersEvents;
  }

  @Override
  public void close() {}

  public static class Builder implements Interceptor.Builder {

    @Override
    public Interceptor build() {
      return new TypeInterceptor();
    }

    @Override
    public void configure(Context context) {}
  }
}
